package start;

import com.bunnymaicai.boot.kafka.autoconfigure.AckMode;
import com.bunnymaicai.boot.kafka.autoconfigure.KafkaListener;
import com.bunnymaicai.boot.kafka.autoconfigure.MessageAck;
import lombok.extern.log4j.Log4j2;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.springframework.stereotype.Component;

/**
 *
 * @author youshang
 * @date 2021/03/01 11:25
 **/
@Log4j2
@Component
public class HelloListener {
    @KafkaListener(value = "test",group = "test-group1",ackMode = AckMode.AUTOMATIC)
    public void hello(ConsumerRecords<Object, Object> consumerRecords, MessageAck messageAck){
        for (ConsumerRecord<Object, Object> consumerRecord : consumerRecords) {

            //只有当 AckMode.MANUAL 的时候手动提交才会生效
//            messageAck.ack(consumerRecord.offset(),consumerRecord.partition());
            log.info(String.format("消费消息成功 : offset :  %s , partition : %s ",consumerRecord.offset(),consumerRecord.partition()));
        }
    }
}
